package com.hz.ruoyiequipment.warning.canal;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.hz.ruoyiequipment.warning.domain.CanalMessage;
import com.hz.ruoyiequipment.warning.service.impl.ThresholdRedisService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.List;
import java.util.Map;

@Component
@Slf4j
public class CanalMessageListener {



    @Autowired
    private ObjectMapper objectMapper; // 用于解析 JSON

    @Autowired
    private ThresholdRedisService thresholdRedisService; // Redis 操作服务

    /**
     * 监听 ny-canal.queue，处理 agr_alert_equip_link 表的变更事件
     *
     * @param message RabbitMQ 消息（JSON 格式）
     * @param channel RabbitMQ 通道
     */
    @RabbitListener(queues = "ny-canal.queue", ackMode = "MANUAL")
    public void handleMessage(Message message, Channel channel) {
        try {
            // 解析 JSON 消息
            CanalMessage canalMessage = objectMapper.readValue(message.getBody(), CanalMessage.class);

            // 判断是否是 agr_alert_equip_link 表的变更
            if ("agr_alert_equip_link".equals(canalMessage.getTable())) {
                // 获取变更类型
                String type = canalMessage.getType(); // 变更类型（INSERT/UPDATE/DELETE）

                // 根据变更类型处理 Redis 缓存
                switch (type) {
                    case "INSERT":
                    case "UPDATE":
                        // 直接重新加载所有数据到 Redis
                        thresholdRedisService.initRedisData();
                        log.info("重新加载 Redis 阈值数据成功，变更类型: {}", type);
                        break;
                    case "DELETE":
                        // 处理删除操作
                        List<Map<String, Object>> data = canalMessage.getData();
                        for (Map<String, Object> row : data) {
                            // 获取外键字段的值
                            Long sourceId = Long.parseLong(row.get("source_id").toString());
                            Long fieldId = Long.parseLong(row.get("field_id").toString());

                            // 构建规则 ID
                            String ruleId = "rule:" + fieldId;

                            // 删除 Redis 缓存
                            thresholdRedisService.deleteThresholdRules(sourceId, ruleId);
                            log.info("删除 Redis 阈值数据成功，设备ID: {}, 规则ID: {}", sourceId, ruleId);
                        }
                        break;
                    default:
                        log.warn("未知的变更类型: {}", type);
                        break;
                }
            }

            // 手动确认消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            log.debug("阈值消息处理成功，消息ID: {}", message.getMessageProperties().getMessageId());
        } catch (Exception e) {
            // 记录异常日志
            log.error("处理阈值消息失败: {}", e.getMessage(), e);

            try {
                // 拒绝消息并重新入队
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                log.warn("阈值消息处理失败，已重新入队，消息ID: {}", message.getMessageProperties().getMessageId());
            } catch (IOException ex) {
                log.error("拒绝阈值消息失败: {}", ex.getMessage(), ex);
            }
        }
    }
}
